use std::{error::Error, time::Duration};
use rabbitmq_consumer_rs::{config::{self, AppConfig, ServerConfig}, logger};
use lapin::{options::*, types::FieldTable, Connection, ConnectionProperties};
use tracing::{debug, error, info, warn};
use futures_lite::StreamExt;
use tokio::{task::JoinSet, sync::mpsc::channel};
use tokio::sync::mpsc::Sender;
use reqwest::header::HeaderMap;
use serde_json::Value;
use serde::{Serialize, Deserialize};
use std::sync::Arc;

#[tokio::main]
async fn main() ->  Result<(), Box<dyn Error>> {

    let config_path = std::env::args().nth(1);
    let path;
    if config_path.is_none() {
        let default_path = "env.json";
        println!("config file path not given. will load from {}",default_path);
        path = format!("{}",default_path);
    }
    else{
        path = config_path.unwrap();
    }
   
    let app_config = config::load_from_file(&path);
    if app_config.is_err() { 
        println!("load config from {} error.{:?}",path,app_config.err());
        println!("the config file content must be like {}",serde_json::to_string(&AppConfig::default()).unwrap());
        return Ok(());
    }
    let config = app_config.unwrap();
    println!("app will start with config {:?}",config);


    let _logger = logger::logger_init(config.logger);
    debug!("already set logger");

    let server_config = config.server;

    if server_config.is_empty() {
        error!("server config is empty. exit!");
        return Ok(());
    }
    let (tx, mut rx) = channel(32); 
    for it in server_config {
        if it.queue_list.is_empty(){
            warn!("server config of queue list is empty. ignored! {:?}",it);            
        }
        else{
            let config = it.clone();
            let sender = tx.clone();
            tokio::spawn(async move{
                let res =  subscriber(sender,config).await;
                if res.is_err() {
                    error!("subscriber error {:?}",res.err());
                }
            });
        }
       
    }
    drop(tx);

    let timeout = config.timeout.unwrap_or(30);
    let retry_times = config.retry_times.unwrap_or(3);
    let retry_interval = config.retry_interval.unwrap_or(20);

    let semaphore  = Arc::new(tokio::sync::Semaphore::new(10));

    while let Some(msg) = rx.recv().await {
        let data: Result<Value, serde_json::Error> = serde_json::from_str(&msg);        
        if data.is_err() {
            error!("receive data is not format json. data:{}. err:{:?}", &msg, data.err());    
            error!("receive content hex is: \n{}\n",hex::encode(&msg));        
        }
        else{
            let data = data.unwrap();
            let url = data.get("url");
            if let Some(url) = url {
                let service = data.get("service");
                if let Some(service) = service {
                    let url = url.as_str().unwrap();
                    let service = service.as_str().unwrap();
                    let params = data.get("params");
                    info!(url = ?url,service = ?service,params = ?params,"callback to service");
                    let url = format!("{}",url);
                    let service = format!("{}",service);
                    let mut call_args = None;
                    if params.is_some() {
                        call_args = Some(serde_json::to_string(params.unwrap())?);
                    }

                    // 限流  防止 CPU暴涨
                    let sig = semaphore.clone(); 
                    let sig_own = sig.acquire_owned().await.unwrap();                    
                    tokio::spawn(async move{
                        callback_handle(url,service,call_args,timeout,retry_interval,retry_times).await;
                        drop(sig_own);
                    });
                } 
                else{
                    error!("receive data does not have service to callback. data:{}",msg);
                }
            }
            else{
                error!("receive data does not have url to callback. data:{}",msg); 
            }
        }        
    }


    Ok(())
}

#[derive(Debug,Clone,Serialize, Deserialize)]
pub struct CallBackData{
    pub service: String,
    pub params: Option<Value>,
}

async fn callback_handle(url:String,service:String,params:Option<String>,timeout:u16,retry_interval:u16,retry_times:u16) {
    
    let mut data = CallBackData{
        service: service.clone(),
        params: None,
    };

    if let Some(args) = &params {
        let json_v = serde_json::from_str(args);
        if json_v.is_err(){
            error!( source =? args,"json data format error. msg: {:?}",json_v.err().unwrap());
            return;
        }
        else{
            data.params = Some(json_v.unwrap());
        }
        
    }
    let times = retry_times + 1;
    let build = reqwest::Client::builder();
    let client = build.connect_timeout(Duration::from_secs(3)).timeout(Duration::from_secs(timeout as u64)).build().unwrap();
        

    for x in 0..times  {
        let mut header = HeaderMap::new();
        // header.insert("User-Agent",   
        // "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36".parse().expect("unsolve the userAgent header value.")
        // );
        header.insert("Content-Type", "application/json; charset=UTF-8".parse().expect("content-type_error"));
    

        let resp = client.post(&url).headers(header).json(&serde_json::json!(data)).send().await;
    
        if resp.is_err() {
            error!(url = ?url,data = ?data, "callback error {:?}",resp.err());
            if retry_times > 0 {
                warn!("will retry {} time after {} secs later.",x + 1,retry_interval);
                tokio::time::sleep(tokio::time::Duration::from_secs(retry_interval as u64)).await;
                //thread::sleep(Duration::from_secs(retry_interval as u64));
            }
        }
        else{
            info!(url = ?url,data = ?data,"callbackOk, resp:{:?}", resp.unwrap().text().await);
            break;
        }
    } 

}

// fn consumer_function(channel: &mut Channel, deliver: protocol::basic::Deliver, headers: protocol::basic::BasicProperties, body: Vec<u8>) {
//     println!("Got a delivery:");
//     println!("Deliver info: {:?}", deliver);
//     println!("Content headers: {:?}", headers);
//     println!("Content body: {:?}", body);
//     channel.basic_ack(deliver.delivery_tag, false);
// }

async fn subscriber(sender:Sender<String>,config:ServerConfig) -> Result<(), Box<dyn std::error::Error>>  {
    if config.queue_list.is_empty() {
        warn!("no queue info in current connection ! {:?}",config);
        return Ok(());
    }

    // amqps://user:pass@hostname:port/vhost
    let addr: String = format!("amqp://{}:{}@{}:{}/{}",
        &config.username,
        &config.password,
        &config.host,
        config.port,
        &config.virtual_host
    );

    let conn = Connection::connect(&addr, ConnectionProperties::default())
    .await
    .expect("connection error");

    info!(addr =? addr, "CONNECTED");

    //receive channel
    let channel = conn.create_channel().await.expect("create_channel");
    info!(state=?conn.status().state());
    info!("will consume");
    let mut set = JoinSet::new();

    for queue in config.queue_list {
        let mut consumer = channel
        .basic_consume(
            &queue.queue,
            &queue.queue,
            BasicConsumeOptions::default(),
            FieldTable::default(),
        )
        .await
        .expect("basic_consume");

        info!(state=?conn.status().state());

        let callback_url = queue.callback_url.unwrap_or_default();
        let isset_url = ! callback_url.is_empty();
        let callback_server = queue.callback_service.unwrap_or_default();
        let isset_server_data = ! callback_server.is_empty();
        let send_handle = sender.clone();
        set.spawn(async move {
            while let Some(delivery) = consumer.next().await {
                // info!(message=?delivery, "received message");
                if let Ok(delivery) = delivery {
                    let msg = std::str::from_utf8(&delivery.data).unwrap();
                    
                    let mut send_data = format!("{}",msg);
                    // 数据类型兼容
                    if isset_url && isset_server_data {
                        send_data = format!("{{\"service\":\"{}\",\"url\":\"{}\",\"params\":[{}]}}",&callback_server,&callback_url,msg);
                    }
        
                    let send_res = send_handle.send(send_data).await;
                    if send_res.is_err() {
                        error!( msg =? msg, "send message error {}",send_res.err().unwrap());
                    }
                    //println!("msg data {}",msg);
                    // let msg = String::from_utf8(delivery.data)?;
                    delivery
                        .ack(BasicAckOptions::default())
                        .await
                        .expect("basic_ack");
                }
            }
        });
    }
    drop(sender);
    while let Some(_) = set.join_next().await {
        info!("some queue consumer exit!");
    }
    // let mut consumer = channel
    //     .basic_consume(
    //         &config.queue,
    //         "my_consumer",
    //         BasicConsumeOptions::default(),
    //         FieldTable::default(),
    //     )
    //     .await
    //     .expect("basic_consume");

    // info!(state=?conn.status().state());

    // while let Some(delivery) = consumer.next().await {
    //     // info!(message=?delivery, "received message");
    //     if let Ok(delivery) = delivery {
    //         let msg = std::str::from_utf8(&delivery.data).unwrap();
            
    //         let mut send_data = format!("{}",msg);
    //         // 数据类型兼容
    //         if config.callback_url.is_some() && config.callback_service.is_some(){
    //             send_data = format!("{{\"service\":\"{}\",\"url\":\"{}\",\"params\":[{}]}}",config.callback_service.clone().unwrap(),config.callback_url.clone().unwrap(),msg);
    //         }

    //         let send_res = sender.send(send_data).await;
    //         if send_res.is_err() {
    //             tracing::error!("send message error {}",send_res.err().unwrap());
    //         }
    //         //println!("msg data {}",msg);
    //         // let msg = String::from_utf8(delivery.data)?;
    //         delivery
    //             .ack(BasicAckOptions::default())
    //             .await
    //             .expect("basic_ack");
    //     }
    // }
    Ok(())
}



// async fn test_fromat_json(){
//     let msg = r#"
//     {"service":"share/service/DeviceBaseService@reportByMqtt","url":"http://account.yeelink.com:30000/inner/amqp/callback","params":[{"serialNo":"03ed000200585591e199","devType":18,"devModel":"","region":"","chanNum":16,"vendor":"HDT-NEUTRAL","P2PVersion":"V1.0.1.2024071218","firmwareInfo":{"upgrade":0,"curVersion":"V2.3.0","curBuildDate":"20240627"},"netInfo":{"type":0,"ethInfo":{"ip":"10.2.0.26","netMask":"255.0.0.0","gateway":"10.0.0.1"}}}]}    
//     "#;

//     let data: Result<Value, serde_json::Error> = serde_json::from_str(msg);
//     if data.is_err(){
//         println!("parse error. error: {:?}",data.err());
//     }
//     else{
//         println!("parse ok. data : {:?}",data.ok().unwrap());
//     }
// }


// #[tokio::test]
// async fn test_format() {
//     let handle  = tokio::spawn(async {
//         test_fromat_json().await;
//     });
//     handle.join().unwrap();
// }

#[test]
fn test_format(){

    let msg = r#"
    {"serialNo":"03ed000200585591e199","devType":18,"devModel":"","region":"","chanNum":16,"vendor":"HDT-NEUTRAL","P2PVersion":"V1.0.1.2024071218","firmwareInfo":{"upgrade":0,"curVersion":"V2.3.0","curBuildDate":"20240627"},"netInfo":{"type":0,"ethInfo":{"ip":"10.2.0.26","netMask":"255.0.0.0","gateway":"10.0.0.1"}}}
    "#;
    let mut send_data = format!("{}",msg);

    let url = "http://account.yeelink.com:30000/inner/amqp/callback".to_owned();
    let service = "share/service/DeviceBaseService@reportByMqtt".to_owned();

    send_data = format!("{{\"service\":\"{}\",\"url\":\"{}\",\"params\":[{}]}}",service,url,send_data);


    let data: Result<Value, serde_json::Error> = serde_json::from_str(&send_data);
    if data.is_err(){
        println!("parse error. error: {:?}",data.err());
    }
    else{
        println!("parse ok. data : {:?}",data.ok().unwrap());
    }

    println!("data hex string is {}",hex::encode(send_data));
}
